import java.util.Properties

import bean.ScalaClass.{After, testUDTFJSON}
import com.alibaba.fastjson.JSON
import fun.lumia.common.Properties.{KAFKA_BOOTSTRAP_SERVERS, KAFKA_GROUP_ID}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.table.sinks.CsvTableSink

object kafkaToHBase {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tabEnv = StreamTableEnvironment.create(env)

    val prop = new Properties()
    prop.setProperty("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS)
    prop.setProperty("group.id", KAFKA_GROUP_ID)
    //    prop.setProperty("auto.offset.reset", "latest")

    val kafkaConsumer = new FlinkKafkaConsumer[String](
      "zhikaiTestUDTF"
      , new SimpleStringSchema()
      , prop)
    kafkaConsumer.setStartFromEarliest()
    val kafkaSource = env.addSource(kafkaConsumer)
    val testUDTFSource = kafkaSource.map(line => {
      val kafkaUDTFJSON = JSON.parseObject(line)
      val table_name = kafkaUDTFJSON.getString("table")
      val op_type = kafkaUDTFJSON.getString("op_type")
      val op_ts = kafkaUDTFJSON.getString("op_ts")
      val current_ts = kafkaUDTFJSON.getString("current_ts")
      val pos = kafkaUDTFJSON.getString("pos")
      val after = kafkaUDTFJSON.getJSONObject("after")
      val ID = after.getString("ID")
      val SRC_CODE = after.getString("SRC_CODE")
      val field0 = after.getString("FIELD0")
      val field1 = after.getString("FIELD1")
      val field2 = after.getString("FIELD2")
      val field3 = after.getString("FIELD3")
      val field4 = after.getString("FIELD4")
      val field5 = after.getString("FIELD5")
      val field6 = after.getString("FIELD6")
      testUDTFJSON(table_name, op_type, op_ts, current_ts, pos, After(ID, SRC_CODE, field0, field1, field2, field3, field4, field5, field6))
    })

    // 注册行转列函数
    tabEnv.registerFunction("hangToLie", new HangToLie())

    tabEnv.registerDataStream("testTable", testUDTFSource)
    val result1 = tabEnv.sqlQuery(
      """
        |select *
        |from testTable,LATERAL TABLE(hangToLie(
        |after.FIELD0
        |,after.FIELD1
        |,after.FIELD2
        |,after.FIELD3
        |,after.FIELD4
        |,after.FIELD5
        |,after.FIELD6
        |)
        |) as T(m_time, num)
        |""".stripMargin)

    //    val result1 = tabEnv.sqlQuery(
    //      """
    //        |select after.ID,after.SRC_CODE from testTable
    //        |""".stripMargin)

    val csvSink = new CsvTableSink("data/sink1")
    val fieldNames = result1.getSchema.getFieldNames
    val fieldTypes = result1.getSchema.getFieldTypes

    tabEnv.registerTableSink("csvSink", fieldNames, fieldTypes, csvSink)

    result1.insertInto("csvSink")

    env.execute()

  }

}
